/**
 * 1) 同步/异步发送：connectionFactory.setUseAsyncSend(true);异步发送通过回调确认是否发送成功
 * 2）消息事务：在seesion创建时确定是否启用事务，如果启用，则commit后消息才能进入Queue或Topic，之前会放在“transaction store”区域中
 * 3）消息持久化：在producer（sender）中设置setDeliveryMode，持久消息在MQ服务器挂掉时依然保存，直到消息被签收
 * 4）消息确认：在seesion创建时确定消费者的消息确认方式（非事务）
 * 5）死信：持久化消息重发送次数超过设定值或过期会被送到死信队列（默认队列名为：ActiveMQ.DLQ）
 * 6）延迟发送：在发送时用message.setLongProperty来设置延迟时间
 */

import org.apache.activemq.*;
import javax.jms.*;
import javax.jms.Message;
import java.time.LocalTime;
import java.util.UUID;

public class AdvancedProducer {
    static QueueConnection connection = null;
    static QueueSession session = null;

    /*默认情况(alwaysSyncSend=false,useAsyncSend=false)，非持久化消息、事务内的消息均采用异步发送；对于持久化消息采用同步发送
              alwaysSyncSend=false，useAsyncSend=true时，持久化消息也采用同步发送
      jms.sendTimeout:发送超时时间，默认等于0，如果jms.sendTimeout>0将会忽略（alwaysSyncSend、useAsyncSend、消息是否持久化）所有的消息都是用同步发送
      异步发送：非持久化消息则没有回执，除非设置connectionFactory.setAlwaysSyncSend(true)。
               持久化消息在未使用事务的时，需设置connectionFactory.setUseAsyncSend(true);在成功完成持久存储操作后，就会向消息生产者发送一个确认信息（也可以设置为不等待回执）
               消息发送者会在发送一定大小的消息后等待服务端进行回执以便确定之前发送的消息是否被正确处理connectionFactory.setProducerWindowSize(102400)
      同步发送：明确指定使用同步发送方式或者在未使用事务的前提下发送持久化的消息（默认）
      生产限流控制：<policyEntry queue=">" producerFlowControl="true" memoryLimit="200mb">，当超过限流时生产者会进入等待状态或者在发送者端直接抛出JMSException
    */
    public static void asyncProducer() throws JMSException {
        ActiveMQConnectionFactory activeMQConnectionFactory =
                new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                        ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616?jms.useAsyncSend=true");

        //开启异步发送
        //activeMQConnectionFactory.setUseAsyncSend(true);
        ActiveMQConnection connection = (ActiveMQConnection) activeMQConnectionFactory.createConnection();

        //开启异步发送
        //connection.setUseAsyncSend(true);

        connection.start();
        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        Queue queue = session.createQueue("asyncQueue");

        ActiveMQMessageProducer activeMQMessageProducer = (ActiveMQMessageProducer) session.createProducer(queue);
        for (int i = 0; i < 5; i++) {
            TextMessage textMessage = session.createTextMessage("Async Message Test----" + i);
            textMessage.setJMSMessageID(UUID.randomUUID() + "----");
            String textMessageId = textMessage.getJMSMessageID();

            //使用ActiveMQMessageProducer的发送消息,可以创建回调,获取发送成功和失败的结果
            activeMQMessageProducer.send(textMessage, new AsyncCallback() {
                @Override
                public void onSuccess() {
                    System.out.println(textMessageId + "Async Message发送成功");
                }

                @Override
                public void onException(JMSException exception) {
                    System.out.println(textMessageId + "Async Message发送失败");
                }
            });
            System.out.println("[Async] Sent:" + textMessage.getText());
        }
        activeMQMessageProducer.close();
        session.close();
        connection.close();
    }


    //事务，带有事务的消息会首先记录在服务器端的“transaction store”区域，并且服务器端会带有redo日志
    //持久化发送，Queue默认是持久化的，Topic默认是非持久化的
    //无论是持久消息，还是非持久消息，如果消息没有对应的消费者，那么activeMQ会认为这些消息无用，直接删除
    //持久消息和非持久消息的区别在于：MQ服务器挂掉的时候，前者不会丢失
    public static void transactionProducer() throws JMSException {
        try {

            QueueConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
            connection = connectionFactory.createQueueConnection();
            connection.start();

            //在事务性会话中消息的确认是自动进行，如果回滚消息会被再次投递
            session = connection.createQueueSession(Boolean.TRUE, Session.SESSION_TRANSACTED);
            Queue queue = session.createQueue("mytranactionqueue");
            QueueSender producer = session.createSender(queue);

            //消息的持久化存储,即使消息服务器出现故障宕机了，消息也不会丢失，直到消息被签收
            //不配置任何的DeliveryMode，activemq.xml中配置persistenceAdapter ---> 持久化，activemq.xml中无persistenceAdapter配置 ---> 不持久化
            //配置DeliveryMode为NON_PERSISTENT，均不持久化
            //配置DeliveryMode为PERSISTENT，均持久化
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);

            for (int i = 0; i < 5; i++) {
                String message = "第" + (i + 1) + "条Transaction Message Test----";
                TextMessage textMessage = session.createTextMessage(message);
                //textMessage.setJMSDeliveryMode(DeliveryMode.PERSISTENT);  单个消息持久化发送
                System.out.println("[Transaction] Sent:" + message);
                producer.send(textMessage);
            }
            //提交消息到服务器队列
            session.commit();
        } catch (Exception e) {
            session.rollback();
        } finally {
            // 关闭释放资源
            if (session != null) {
                session.close();
            }
            if (connection != null) {
                connection.close();
            }
        }

    }

    //持久化订阅
    public static void topicProducer() throws JMSException {
        try {
            TopicConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
            TopicConnection connection = connectionFactory.createTopicConnection();
//            connection.setClientID("myClientId");
//            connection.start();
            TopicSession session = connection.createTopicSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

            Topic topic = session.createTopic("mytopic");
            // 6、使用Session对象创建一个Producer对象。
            MessageProducer producer = session.createProducer(topic);
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            String message = "Durable Message Test----"+LocalTime.now();
            TextMessage textMessage = session.createTextMessage(message);
            System.out.println("[Durable] Sent:" + message);
            producer.send(textMessage);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭释放资源
            if (session != null) {
                session.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
    }

    //消息接收确认
    public static void ackProducer() throws JMSException {
        try {
            QueueConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
            connection = connectionFactory.createQueueConnection();
            connection.start();

            //Session.AUTO_ACKNOWLEDGE（1）:表示session会自动确认所接收到的消息，但消费端不一定是向服务端一条一条确认。
            //Session.CLIENT_ACKNOWLEDGE（2）:手动批量确认所接收到的消息。
            //Session.DUPS_OK_ACKNOWLEDGE（3）:使得session将“惰性”地确认消息，即不会立即确认消息，这样可能导致消息重复投递。
            //Session.INDIVIDUAL_ACKNOWLEDGE（4）:手动单条确认所接收到的消息
            session = connection.createQueueSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);
            Queue queue = session.createQueue("myackqueue");
            QueueSender producer = session.createSender(queue);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            for (int i = 0; i < 5; i++) {
                String message = "Ack Message Test----"+ (i + 1);
                TextMessage textMessage = session.createTextMessage(message);
                System.out.println("[ACK] Sent:" + message);
                producer.send(textMessage);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭释放资源
            if (session != null) {
                //提交消息到服务器队列
                session.close();
            }
            if (connection != null) {
                connection.close();
            }
        }

    }

    /*死信，消息重发超过规定次数（默认6次），该消息会被送入死信队列。下列情况会导致消息重发（默认间隔时间为1s）
      1）Client用了transaction且在session中调用了rollback或者commit之前session被关闭
      2）Client在CLIENT_ACKNOWLEDGE的模式下，session中调用了recover或者ack之前session被关闭
        消息重发参数可在RedeliveryPolicy中设置，然后connectionFactory.setRedeliveryPolicy即可
        默认情况下持久消息过期或超过重发次数都会被送到死信队列，<sharedDeadLetterStratgy processExpired = "true"/>
        默认情况非持久消息过期或超过重发次数会直接删除，而不会送到死信队列，<sharedDeadLetterStrategy processNonPersistent="false" />
    */
    public static void deadProducer() throws Exception {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setInitialRedeliveryDelay(0);
        redeliveryPolicy.setRedeliveryDelay(1000);
        redeliveryPolicy.setUseExponentialBackOff(false);
        redeliveryPolicy.setMaximumRedeliveries(2);    // 修改重发次数为2次
        connectionFactory.setRedeliveryPolicy(redeliveryPolicy);
        ActiveMQConnection mqConnection = (ActiveMQConnection) connectionFactory.createQueueConnection();
        mqConnection.start();
        session = mqConnection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("deadQueue");
        QueueSender producer = session.createSender(queue);
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        QueueReceiver consumer = session.createReceiver(queue);
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println("[deadQueue]received：" + textMessage.getText() +"----" + LocalTime.now());
                    session.rollback();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

        TextMessage textMessage = session.createTextMessage("DeadMesssage-----" + LocalTime.now());
        producer.send(textMessage);
        System.out.println("[deadQueue]sent：" +textMessage.getText());
        session.commit();
        System.in.read();
        session.close();

        mqConnection.close();
    }


    //延时发送，activemq.xml中设置 <broker xmlns.... schedulerSupport="true">;message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
    public static void delayProducer() throws Exception {
        long delay = 2 * 1000;
        long perid = 4 * 1000;
        int repeat = 5;

        QueueConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");

        connection = connectionFactory.createQueueConnection();
        connection.start();

        session = connection.createQueueSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);
        Queue queue = session.createQueue("delayQueue");
        QueueSender producer = session.createSender(queue);

        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        TextMessage textMessage = session.createTextMessage("Delay Message Test" + LocalTime.now());
        // 消息每过 3 秒投递，每 4 秒重复投递一次 ，一共重复投递 5+1次
        textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
        textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, perid);
        textMessage.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);
        producer.send(textMessage);
        session.close();
        connection.close();
    }

}
